package yanglib.application.metadata.client;

import yanglib.application.metadata.MetadataMembership;
import yanglib.application.metadata.message.GetMessage;
import yanglib.application.metadata.message.MetadataMessageDeserializer;
import yanglib.application.metadata.message.PutMessage;
import yanglib.application.metadata.server.ServerMessageHandler;
import yanglib.network.Network;
import yanglib.network.nio.NIONetwork;
import yanglib.network.tcp.TcpNetwork;
import yanglib.session.*;

import java.io.File;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.*;

/**
 * Created with IntelliJ IDEA.
 * User: user
 * Date: 9/6/14
 * Time: 9:25 PM
 * To change this template use File | Settings | File Templates.
 */
public class MetadataClient implements Callback {

    private ClientSession session = null;
    private Network network = null;
    private MetadataMembership membership = null;
    private int selfID;
    private int writeCount = 2;
    private int readCount = 1;

    public MetadataClient(int selfID, String configFile){
        this.selfID = selfID;
        membership = new MetadataMembership(new File(configFile));
        session = new ClientSession(selfID, true);
        MetadataMessageDeserializer deserializer = new MetadataMessageDeserializer();
        network = new NIONetwork(membership, session, deserializer, selfID);
        session.setNetwork(network);
        //targets.add(3);
    }

    private HashSet<Integer> getWriteTargets(byte[] key){
        int hash = 0;
        for(int i=0; i<key.length; i++)
            hash += key[i];
        int serverCount = membership.getRoleCount("server");
        int current = hash % serverCount;
        HashSet<Integer> ret = new HashSet<Integer>();
        for(int i=0; i<writeCount; i++){
            int id = Integer.parseInt(membership.getProperty("server", i, "id"));
            ret.add(id);
            current = (current + 1) % serverCount;
        }
        return ret;
    }

    private HashSet<Integer> getReadTargets(byte[] key){
        int hash = 0;
        for(int i=0; i<key.length; i++)
            hash += key[i];
        int serverCount = membership.getRoleCount("server");
        int current = hash % serverCount;
        HashSet<Integer> ret = new HashSet<Integer>();
        for(int i=0; i<readCount; i++){
            int id = Integer.parseInt(membership.getProperty("server", i, "id"));
            ret.add(id);
            current = (current + 1) % serverCount;
        }
        return ret;
    }

    public void put(byte[] key, byte[]value) throws IOException {
        PutMessage putMessage = new PutMessage(selfID, key, value);
        session.submitRequest(putMessage, this, getWriteTargets(key), 10000);
    }

    public void get(byte[] key) throws IOException{
        GetMessage getMessage = new GetMessage(selfID, key);
        session.submitRequest(getMessage, this, getReadTargets(key), 10000);
    }

    public boolean handleReply(PendingRequest pRequest, Reply reply){
        //System.out.println("Got a reply "+reply);
        pRequest.addReply(reply);
        if(pRequest.getRequest() instanceof PutMessage && pRequest.getReplies().size() >= writeCount)
            return true;
        else if(pRequest.getRequest() instanceof GetMessage && pRequest.getReplies().size() >= readCount)
            return true;
        else
            return false;
    }

    public void complete(PendingRequest pRequest){
        //System.out.println(pRequest.getRequest()+" completed");
    }

    public void handleError(PendingRequest pRequest, int target, Exception e){
        //System.out.println("Got an error from node "+target);
        e.printStackTrace();
    }

    public void timeout(PendingRequest pRequest){
        //System.out.println(pRequest+" timeout");
        Set<Integer> resendTargets = pRequest.getUnRepliedTargets();
        pRequest.resend(resendTargets, 20000);
    }

    public static void main(String[] args) throws Exception {
        if(args.length != 3){
            System.out.println("java MetadataClient <id> <configFile> <requestCount>");
            return;
        }
        int id = Integer.parseInt(args[0]);
        String configFile = args[1];
        long count = Long.parseLong(args[2]);

        MetadataClient client = new MetadataClient(id, configFile);
        Random r = new Random();
        long startTime = System.currentTimeMillis();
        for(long i=0; i<count; i++){
            byte[] keyBytes = new byte[8];
            byte[] valueBytes = new byte[4096];
            r.nextBytes(keyBytes);
            r.nextBytes(valueBytes);
            client.put(keyBytes, valueBytes);
        }
        System.out.println("throughput = " + count * 1000L / (System.currentTimeMillis() - startTime));
        System.exit(0);
    }

}
